package com.lsh.oms.kafka.service.SpringKafka;

import com.alibaba.fastjson.JSON;
import com.lsh.oms.core.exception.BusinessException;
import com.lsh.oms.core.model.bills.MessageBean;
import com.lsh.oms.core.mq.MessageProducer;
import com.lsh.oms.core.service.bills.HandlerKafkaMessage;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * Project Name: lsh-oms
 *
 * @author peter
 * @date 17/4/11
 * 北京链商电子商务有限公司
 * Package
 * desc:
 */
@Component
public class KafkaReceiver {

    private static Logger logger = LoggerFactory.getLogger(KafkaReceiver.class);

    @Resource
    private HandlerKafkaMessage handler;

    @Resource(name = "kafka.message.producer")
    private MessageProducer producer;

    @KafkaListener(topics = {"${topic.name.order}", "${topic.name.receipt}", "${topic.name.return}"}, containerFactory = "concurrentKafkaListenerContainerFactory")
    public void processMessage(ConsumerRecord<String, String> record, Acknowledgment ack) {
        String topic = record.topic();
        String content = record.value();
        long offset = record.offset();
        logger.info("pull kafka message: {} on topic: {}, offset: {}", new Object[]{content, topic, offset});
        beautify(content).ifPresent(message -> {
            if (message.shouldConsume()) {
                try {
                    handler.handlerKafkaMessage(message);
                } catch (BusinessException e) {
                    postpone(topic, message);
                    logger.error("业务异常, code: " + e.getCode() + ", exception: " + e.getMessage() + ", message content: " + content, e);
                } catch (Exception e) {
                    postpone(topic, message);
                    logger.error("系统异常, exception: " + e.getMessage() + ", message content: " + content, e);
                }
            } else {
                putBack(topic, message);
            }
        });
        commit(ack, topic, offset);
    }

    private Optional<MessageBean> beautify(String message) {
        try {
            return Optional.ofNullable(JSON.parseObject(message, MessageBean.class));
        } catch (Exception e) {
            logger.error("exception parse message: " + message + " to object, exception message: " + e.getMessage(), e);
        }
        return Optional.empty();
    }

    private void commit(Acknowledgment ack, String topic, long offset) {
        try {
            ack.acknowledge();
            logger.info("manual commit success, topic: {}, offset: {}", topic, offset);
        } catch (Exception e) {
            logger.error("manual commit failed, topic: " + topic + ", offset: " + offset, e);
        }
    }

    private void putBack(String topic, MessageBean message) {
        logger.info("put back message, should consume at: {}", message.getScheduledAt());
        producer.send(topic, message);
    }

    private void postpone(String topic, MessageBean message) {
        message.setScheduledAt(new DateTime().plusMinutes(1).getMillis());
        logger.info("postpone message, should consume at: {}", message.getScheduledAt());
        producer.send(topic, message);
    }
}
